package com.yunhe.realtime.websocket;

import com.yunhe.common.constant.CacheConsts;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.realtime.service.data.RealtimeService;
import com.yunhe.common.websocket.WebsocketRealtimeHandler;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 自定义实时数据WebSocket处理类
 * @author liuronglei
 */
public class CustomHandler extends WebsocketRealtimeHandler {
    @Autowired
    private RealtimeService realtimeService;

    @Override
    public String getPointField() {
        return "key";
    }

    @Override
    public void beforeSend(WebSocketSession session, Object content) {

    }

    @Override
    public void afterEstablished(WebSocketSession session) {

    }

    @Override
    public void afterClosed(WebSocketSession session, CloseStatus status) {

    }

    @Override
    public void handleMessage(WebSocketSession session, TextMessage message, List<String> points) {
        Map<Object, Object> result = new HashMap<>();
        for (String point : points) {
            Map onResult = realtimeService.getMap(CacheConsts.KEY.CUSTOM_REALTIME.value(point));
            result.put(point, onResult);
        }
        sendMessage(result, session, points);
    }

    /**
     * 监听kafka中的自定义实时值消息
     * @param record 记录
     */
    @KafkaListener(topics = KafkaConsts.Topic.CustomRealtime, groupId = KafkaConsts.Group.RandomGroupId)
    public void customRealtimeListener(ConsumerRecord<?, String> record) {
        String key = record.value();
        Map onResult = realtimeService.getMap(CacheConsts.KEY.CUSTOM_REALTIME.value(key));
        Map<Object, Object> result = new HashMap<>();
        result.put(key, onResult);
        sendMessage(result, null, Collections.singletonList(key));
    }

}